package com.swak.reactivex.cluster;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.reactivex.cluster.impl.ClusterHandler;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.utils.Lists;
import com.swak.utils.Maps;

/**
 * Members 管理所有的成员.
 * 
 * @author DELL
 */
public class Members {

	private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);

	private Map<ID, Member> members = Maps.newConcurrentMap();
	private ClusterHandler chanager;

	public Members(ClusterHandler chanager) {
		this.chanager = chanager;
	}

	public int count() {
		return this.members.size();
	}

	public Member check(ID id) {
		Member exist = id != null ? this.members.get(id) : null;
		if (exist != null && !(exist.id().getUrl().getAddress().equals(id.getUrl().getAddress()))) {
			throw new MemberExistException("Member Id Is Conflict：" + id);
		}
		return exist;
	}

	public Member lookup(ID id) {
		return id != null ? this.members.get(id) : null;
	}

	public List<Member> members() {
		return Lists.newArrayList(this.members.values());
	}

	public Members register(Member member) {

		this.members.put(member.id(), member);

		if (LOGGER.isDebugEnabled() && (member.channels().size() > 0 || member.isThis())) {
			LOGGER.debug(this.toString());
		}

		if (this.chanager != null) {
			this.chanager.memberShipChanaged(member, Lists.newArrayList(this.members.values()), true);
		}

		return this;
	}

	public Members unregister(Member member) {

		try {
			this.members.remove(member.id());
			member.close();
		} catch (Exception e) {
			LOGGER.error("un register member [" + member.toString() + "] error.", e);
		}

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug(this.toString());
		}

		if (this.chanager != null) {
			this.chanager.memberShipChanaged(member, Lists.newArrayList(this.members.values()), false);
		}

		return this;
	}

	public Members register(Member member, ExchangeChannel channel) {
		member.channels().add(channel);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug(this.toString());
		}
		return this;
	}

	public void close() {
		this.members.values().stream().forEach(member -> {
			this.unregister(member);
		});
	}

	@Override
	public String toString() {
		StringBuilder out = new StringBuilder();
		out.append("\nMembers {size:").append(this.members.size()).append("}[").append("\n");
		int i = 1;
		for (Member member : this.members.values()) {
			out.append("  ").append(i++).append(".\t").append(member.toString()).append("\n");
		}
		out.append("]");
		return out.toString();
	}

	/**
	 * 节点列表变动时；
	 * 
	 * @author DELL
	 */
	@FunctionalInterface
	public interface MemberShipChanager {

		/**
		 * 变动事件
		 * 
		 * @param member 变动的节点
		 * @param members 所有节点
		 * @param add true： 增加节点
		 */
		void chanage(Member member, List<Member> members, boolean add);
	}
}